package com.zisu.rabbitconsumer.component;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Consumer;
import com.zisu.rabbitconsumer.Entity.User;
import com.zisu.rabbitconsumer.test.MyConsumer;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.ConsumerTagStrategy;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@Component
public class RabbitReceive {

    /**
     *  @Description  组合监听使用
     *  @Param
     *  @Return
     *  @Author caixiang
     *  @Date 2020/6/17  10:43
     **/
    // topic 模式
    // 1.这个注解是 监听某个队列 的；；2.如果交换机和 队列不存在那么 在启动的时候回去 创建它；
    //客户端 监听队列 ，这个队列并不需要事先创建，意思就是 我客户端就依据这个key 来接受数据
    @RabbitListener(
        bindings = @QueueBinding(
                value = @Queue(value="queue1",durable = "true"),
                exchange = @Exchange(name = "exchange1",durable = "true",type = "topic",ignoreDeclarationExceptions = "true"),

                key = "springboot.*"
        )
    )
    //只要有消息过来都会执行这个函数
    @RabbitHandler
    public void onMessage(Message<?> message, Channel channel)throws Exception{
        //1.收到消息后  业务端对message 进行处理
        System.out.println("===============================");
        //header是固定的(每个事件)
        System.out.println("端口:8004 消费头："+message.getHeaders().get("attr1")+"|"+message.getHeaders().get("attr2"));
        //Body 每个事件不同。
        System.out.println("端口:8004 消费体："+message.getPayload());
        Thread.sleep(5000);
        //2.处理成功后 获取deliveryTag 并进行手工的ACK操作
        //因为我们在配置文件（yml）中 配置了手工签收消息，所以我们代码中处理完消息要签收（手工ack）
        //当生产者 生产出消息后 就会自动在message 的Header 上 赋上 DELIVERY_TAG
        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        //multiple：为了减少网络流量，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
        //如果不去Ack 那么下次来 还会把 这个消息再消费一次。
        //参数1：deliveryTag:该消息的index； 参数2：multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息。


        //消费成功了 肯定不用重回队列了，当然也可以回队列
        channel.basicAck(deliveryTag,false);

        //对于消费失败的数据可以重回队列
        //channel.basicNack(deliveryTag,false,true);
    }


    //Direct模式
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value="eq1_request_queue",durable = "true"),
                    exchange = @Exchange(name = "es1_exchange",durable = "true",type = ExchangeTypes.DIRECT,ignoreDeclarationExceptions = "true"),
                    //这里的key 就是 bingingKey ，因为是Direct模式所以，只有bindingKey（交换机和队列的）和 routingKey（消息的） 完全匹配后，交换机才会把这条消息投递到这条队列
                    key = "eq1_request"
            )
    )
    // direct 模式
    //只要有消息过来都会执行这个函数
    @RabbitHandler
    public void es1ExchangeEq1(Message<?> message, Channel channel)throws Exception{
        //1.收到消息后  业务端对message 进行处理
        System.out.println("==============eq1-request=================");
        System.out.println("端口:8004 消费体："+message.getPayload());
        System.out.println("端口:8004 消费头："+message.getHeaders().get("attr1")+"|"+message.getHeaders().get("attr2"));
        Thread.sleep(5000);
        //2.处理成功后 获取deliveryTag 并进行手工的ACK操作
        //因为我们在配置文件（yml）中 配置了手工签收消息，所以我们代码中处理完消息要签收（手工ack）
        //当生产者 生产出消息后 就会自动在message 的Header 上 赋上 DELIVERY_TAG
        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);

        //multiple：为了减少网络流量，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
        //如果不去Ack 那么下次来 还会把 这个消息再消费一次。
        //参数1：deliveryTag:该消息的index； 参数2：multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息。
        //消费成功了 肯定不用重回队列了，当然也可以回队列
        channel.basicAck(deliveryTag,false);

        //对于消费失败的数据可以重回队列
        //消息退回 (可以在可视化界面看到)
        //批量退回 退回之后重回消息队列 true（最好不要重回队列 因为一次失败 第二次 大概率 也失败）；  false的话就是丢弃这条信息，如果配置了死信队列，那这条消息会进入死信队列
        //如果一直重新 消费一直异常 就会报错 ConditionalRejectingErrorHandler : Execution of Rabbit message listener failed.
        //channel.basicNack(deliveryTag,false,true);
    }


    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value="eq1_response_queue",durable = "true"),
                    exchange = @Exchange(name = "es1_exchange",durable = "true",type = ExchangeTypes.DIRECT,ignoreDeclarationExceptions = "true"),
                    key = "eq1_response"
            )
    )
    @RabbitHandler
    public void es1ExchangeResponseEq1(Message<?> message, Channel channel)throws Exception{
        //1.收到消息后  业务端对message 进行处理
        System.out.println("==============eq1-response=================");
        System.out.println(new Date() +";;端口:8004 消费头："+message.getHeaders().get("attr1")+"|"+message.getHeaders().get("attr2"));



        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag,false);
    }




    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value="eq2_request_queue",durable = "true"),
                    exchange = @Exchange(name = "es1_exchange",durable = "true",type = "direct",ignoreDeclarationExceptions = "true"),
                    key = "eq2_request"
            )
    )
    // direct 模式
    //只要有消息过来都会执行这个函数
    @RabbitHandler
    public void es1ExchangeEq2(Message<?> message, Channel channel)throws Exception{
        //1.收到消息后  业务端对message 进行处理
        System.out.println("==============eq2-request=================");
        System.out.println("端口:8004 消费头："+message.getHeaders().get("attr1")+"|"+message.getHeaders().get("attr2"));
        byte[] bytes =(byte[])message.getPayload();
        ObjectMapper objectMapper = new ObjectMapper();
        User user = objectMapper.readValue(bytes, User.class);

        //System.out.println("端口:8004 消费体："+message.getPayload());
        Thread.sleep(5000);
        //2.处理成功后 获取deliveryTag 并进行手工的ACK操作
        //因为我们在配置文件（yml）中 配置了手工签收消息，所以我们代码中处理完消息要签收（手工ack）
        //当生产者 生产出消息后 就会自动在message 的Header 上 赋上 DELIVERY_TAG
        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);

        //multiple：为了减少网络流量，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
        //如果不去Ack 那么下次来 还会把 这个消息再消费一次。
        //参数1：deliveryTag:该消息的index； 参数2：multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息。
        //消费成功了 肯定不用重回队列了，当然也可以回队列
        channel.basicAck(deliveryTag,false);

        //对于消费失败的数据可以重回队列
        //channel.basicNack(deliveryTag,false,true);
    }
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value="eq2_response_queue",durable = "true"),
                    exchange = @Exchange(name = "es1_exchange",durable = "true",type = ExchangeTypes.DIRECT,ignoreDeclarationExceptions = "true"),
                    key = "eq2_response"
            )
    )
    @RabbitHandler
    public void es1ExchangeResponseEq2(Message<?> message, Channel channel)throws Exception{
        //1.收到消息后  业务端对message 进行处理
        System.out.println("==============eq2-response=================,priority"+message.getHeaders().get("priority"));
        System.out.println("端口:8004 消费体："+message.getPayload());
        System.out.println("端口:8004 消费头："+message.getHeaders().get("attr1")+"|"+message.getHeaders().get("attr2"));
        Thread.sleep(5000);
        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag,false);
    }


    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value="remote_command_request_queue",durable = "true"),
                    exchange = @Exchange(name = "rc_exchange",durable = "true",type = "topic",ignoreDeclarationExceptions = "true"),
                    //这个key就是binding key，因为是topic模式 所以key是一个表达式，交换机会依据 生产者产出的消息里面的 routingkey 和 这个bingKey进行匹配然后投递到相应队列里面去
                    //如：这里rc_exchange交换机 会把所有满足 RC.#这个表达式 的routringkey消息 全部转发到remote_command_request_queue 这个队列里面去。
                    key = "RC.#"
            )
    )
    // TOPIC 模式
    //只要有消息过来都会执行这个函数
    @RabbitHandler
    public void remoteCommand(Message<?> message, Channel channel)throws Exception{
        //1.收到消息后  业务端对message 进行处理
        System.out.println("==============remote_command-rc=================");
        System.out.println("端口:8004 消费头："+message.getHeaders().get("attr1")+"|"+message.getHeaders().get("attr2"));
        byte[] bytes =(byte[])message.getPayload();
        ObjectMapper objectMapper = new ObjectMapper();
        User user = objectMapper.readValue(bytes, User.class);
        Consumer defaultConsumer = channel.getDefaultConsumer();

        //System.out.println("端口:8004 消费体："+message.getPayload());
        Thread.sleep(5000);
        //2.处理成功后 获取deliveryTag 并进行手工的ACK操作
        //因为我们在配置文件（yml）中 配置了手工签收消息，所以我们代码中处理完消息要签收（手工ack）
        //当生产者 生产出消息后 就会自动在message 的Header 上 赋上 DELIVERY_TAG
        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);

        //multiple：为了减少网络流量，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
        //如果不去Ack 那么下次来 还会把 这个消息再消费一次。
        //参数1：deliveryTag:该消息的index； 参数2：multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息。
        //消费成功了 肯定不用重回队列了，当然也可以回队列
        channel.basicAck(deliveryTag,false);

        //对于消费失败的数据可以重回队列
        //channel.basicNack(deliveryTag,false,true);
    }


    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value="remote_command_response_queue",durable = "true"),
                    exchange = @Exchange(name = "rc_exchange",durable = "true",type = "topic",ignoreDeclarationExceptions = "true"),
                    key = "ACK.#"
            )
    )
    // TOPIC 模式
    //只要有消息过来都会执行这个函数
    @RabbitHandler
    public void remoteCommand2(Message<?> message, Channel channel)throws Exception{
        System.out.println("==============remote_command-ack=================");
        System.out.println("端口:8004 消费头："+message.getHeaders().get("attr1")+"|"+message.getHeaders().get("attr2"));
        byte[] bytes =(byte[])message.getPayload();
        ObjectMapper objectMapper = new ObjectMapper();
        User user = objectMapper.readValue(bytes, User.class);
        Thread.sleep(5000);
        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);


        try {
            channel.basicAck(deliveryTag,false);
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value="remote_command_response_queue",durable = "true"),
                    exchange = @Exchange(name = "rc_exchange",durable = "true",type = "topic",ignoreDeclarationExceptions = "true"),
                    key = "RC.#"
            )
    )
    // TOPIC 模式
    //只要有消息过来都会执行这个函数
    @RabbitHandler
    public void remoteCommandForTest(Message<?> message, Channel channel)throws Exception{
        System.out.println("==============remote_command-Remote=================");
        System.out.println("端口:8004 消费头："+message.getHeaders().get("attr1")+"|"+message.getHeaders().get("attr2"));
        byte[] bytes =(byte[])message.getPayload();
        ObjectMapper objectMapper = new ObjectMapper();
        User user = objectMapper.readValue(bytes, User.class);
        Thread.sleep(5000);
        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);

        try {
            channel.basicAck(deliveryTag,false);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
    //#############################################################################################
    @Resource
    private ConnectionFactory connectionFactory;

    private int a = 1;



    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value="myhelloQueue",durable = "true",
                            arguments = {
                                    //1. 死信队列Demo
                                    @Argument(name = "x-dead-letter-exchange", value = "es1_dead_message_exchange"),
                                    //可以为死信 指定routingKey，如果不指定routingKey的话 那么 会用原routingKey也就是 eq5_request
                                    @Argument(name = "x-dead-letter-routing-key", value = "es1_eq1_dead_message"),
                                    //这里设置的优先级 是这个队列承认的 最高优先级值，也就是说超过10 就 都一样。

                                    //2.优先级队列Demo
                                    @Argument(name = "x-max-priority", value = "10",type = "java.lang.Integer")
                                    //设置队列最大长度
                                    //@Argument(name = "x-max-length", value = "500", type = "java.lang.Integer")
                            }
                    ),
                    exchange = @Exchange(name = "myhelloExchange",durable = "true",type = ExchangeTypes.DIRECT,ignoreDeclarationExceptions = "true"),
                    key = "myhelloRouting"
            )
            //一个 @RabbitListener 就包含一个 containerFactory
//            ,
//            containerFactory =
    )
    @RabbitHandler
    public void testforHAExchange22(Message<?> message, Channel channel)throws Exception{

        //1.收到消息后  业务端对message 进行处理
        System.out.println();
        System.out.println("==============forHAFailover=================");
        Thread.sleep(5000);
        Long deliveryTag = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTag,false);
    }



}